/**
 * 
 */
package com.pub.proxy;

import java.net.InetSocketAddress;
import java.net.Proxy;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;

import org.apache.log4j.Logger;
import org.apache.log4j.xml.DOMConfigurator;

import com.pub.db.DBConnectionManager;




/**
 * @author iling
 *
 */
public class ProxyManager 
{
	private final int INT_WAIT_TIMEOUT 			= 30; 	//--> unit: second, default timeout for fetch Proxy
	private final int INT_PROXY_POOL_SIZE		= 100000; //--> Proxy cache size
	private final int INT_INVALID_PROXY_LIMIT 	= 2; 	//--> if proxy is invalid, allow fail count.
	
	private final String STR_CONFIG_FILES_PATH_NAME_FOR_SYSTEM_ENV = "LOCAL_PUB_CONFIG_DIR";
	
	private static boolean isFetch;
	
	private static DBConnectionManager dbm;
	private static Logger logger;
	
	private static HashSet<ProxyElement> hsCheckOutProxy;
	
	private static Hashtable<String, Integer> htClientProxy;
	private static Hashtable<String, ProxyElement> htProxyPool;
	
	protected static String STRConfigFileDir, DBTag;
	protected static ProxyManager proxyInstance; 	//-- only instance

	private BlockingQueue<ProxyElement> bkqProxyQueue, bkgInvalidProxyQueue;
	
	static
	{
		htClientProxy 	= new Hashtable<String, Integer>(50);
		logger 			= Logger.getLogger (ProxyManager.class.getName());
	}
	
	/**
	 * @param args
	 */
	public static void main(String[] args) 
	{
		ProxyManager service = ProxyManager.getInstance(ProxyManager.class.getName(), "crawler", "/home/workspace/crawler/configs/");
		
		try {
			Thread.sleep(10000);

			ProxyElement proxy;
			
			System.out.println("proxy element size : " + service.getProxyElementSize());
			
			for(int i = 0; i < service.getProxyElementSize(); i++){
				proxy = service.getProxy();
				
				service.freeProxy(proxy);
				
				System.out.println("getValidCount : " + proxy.getValidCount());
				System.out.println("getInvalidCount : " + proxy.getInvalidCount());
			}
			
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/* return only instance
	 * @ parameter 	 client Class Name
	 */
	public static synchronized ProxyManager getInstance(String clientClassName) 
	{
    	if (proxyInstance == null) {
    		proxyInstance = new ProxyManager(clientClassName, "log4j.xml");
		}
		
    	if (proxyInstance != null) {
	    	if(htClientProxy.containsKey(clientClassName)){
	    		htClientProxy.put(clientClassName, htClientProxy.get(clientClassName) + 1);
	    	}else{
	    		htClientProxy.put(clientClassName, 1);
	    	}
    	}
    	
		return proxyInstance;
	}
	
	/* return only instance
	 * @ parameter 	 client Class Name
	 */
	public static synchronized ProxyManager getInstance(String clientClassName, String DBTag) 
	{
		ProxyManager.DBTag = DBTag;
    	
		if (proxyInstance == null) {
    		proxyInstance = new ProxyManager(clientClassName, "log4j.xml");
		}
		
    	if (proxyInstance != null) {
	    	if(htClientProxy.containsKey(clientClassName)){
	    		htClientProxy.put(clientClassName, htClientProxy.get(clientClassName) + 1);
	    	}else{
	    		htClientProxy.put(clientClassName, 1);
	    	}
    	}
    	
		return proxyInstance;
	}
	
	/* return only instance
	 * @ parameter 	 client Class Name
	 */
	public static synchronized ProxyManager getInstance(String clientClassName, String DBTag, String XMLconfigPath) 
	{
		ProxyManager.DBTag = DBTag;
    	
		if (proxyInstance == null) {
    		proxyInstance = new ProxyManager(clientClassName, XMLconfigPath);
		}
		
    	if (proxyInstance != null) {
	    	if(htClientProxy.containsKey(clientClassName)){
	    		htClientProxy.put(clientClassName, htClientProxy.get(clientClassName) + 1);
	    	}else{
	    		htClientProxy.put(clientClassName, 1);
	    	}
    	}
    	
		return proxyInstance;
	}
	
	/**
	 * 
	 */
	protected ProxyManager() 
	{
		init();
		timerControl(300000); //-- timer，unit：millisecond(default: 5 minutes)
	}

	/**
	 * 
	 */
	protected ProxyManager(String strClientClassName, String XMLconfigPath) 
	{
		STRConfigFileDir = System.getenv(STR_CONFIG_FILES_PATH_NAME_FOR_SYSTEM_ENV);
	    
		if(XMLconfigPath != null){
			STRConfigFileDir = XMLconfigPath;
		}
		
	    if(STRConfigFileDir == null){
	    	System.out.println("can't find CPS_LOCAL_CONFIG_DIR from system env.");
	    	System.out.println("Proxy Manager initialize failed.");
	    }else{
	    	System.out.println("initialize log4j from " + STRConfigFileDir + "log4j.xml");
			
		    init();
		    timerControl(300000); //-- timer，unit：millisecond(default: 5 minutes)
	    }
	}

	private void init()
	{
		isFetch = false;
		
		DOMConfigurator.configure(STRConfigFileDir + "log4j.xml");

		bkqProxyQueue 			= new PriorityBlockingQueue<ProxyElement>(INT_PROXY_POOL_SIZE);
		bkgInvalidProxyQueue	= new ArrayBlockingQueue<ProxyElement>(INT_PROXY_POOL_SIZE);
		htClientProxy 			= new Hashtable<String, Integer>(INT_PROXY_POOL_SIZE);
		htProxyPool				= new Hashtable<String, ProxyElement>(INT_PROXY_POOL_SIZE);
		
		hsCheckOutProxy 		= new HashSet<ProxyElement>(INT_PROXY_POOL_SIZE);
		
		dbm 	= DBConnectionManager.getInstance(ProxyManager.class.getName(), STRConfigFileDir);

	}
	
	
	
	public synchronized int getProxyElementSize()
	{
		return bkqProxyQueue.size();
	}
	
	/* fetch usable Proxy.
 	 * @ return Connection or null(not usable connection)
	 */
	public synchronized ProxyElement getProxy() throws NullProxyException
	{
		ProxyElement proxy = null;
		
		try {
			proxy = this.bkqProxyQueue.poll(this.INT_WAIT_TIMEOUT, TimeUnit.SECONDS);
		
			hsCheckOutProxy.add(proxy);
		} catch (InterruptedException e) {
			logger.warn("interrupted while waiting fetch Proxy.");
			
			throw new NullProxyException();
		}
		
		if(proxy == null)
			throw new NullProxyException();
		
		return proxy;
	}
	
	/* reclaim connection
	 * @ param con 		used connection
	 */
	public synchronized void freeProxy(Proxy proxy) 
	{
		if(proxy == null)
			return;
		
		if(!hsCheckOutProxy.contains(proxy)){
			logger.warn("can't find this proxy for ProxyManager.");
			
			return;
		}else{
			hsCheckOutProxy.remove(proxy);
		}

		if(proxy instanceof ProxyElement){
			ProxyElement proxyElement = (ProxyElement)proxy;
			
			try {
				if(proxyElement.getInvalidCount() < INT_INVALID_PROXY_LIMIT)
					this.bkqProxyQueue.offer(proxyElement, this.INT_WAIT_TIMEOUT, TimeUnit.SECONDS);
				else{
					bkgInvalidProxyQueue.offer(proxyElement, this.INT_WAIT_TIMEOUT, TimeUnit.SECONDS);
					
					if(htProxyPool.containsKey(proxyElement.getHost()))
						htProxyPool.remove(proxyElement.getHost());
				}
			} catch (Exception e) {
				if(htProxyPool.containsKey(proxyElement.getHost()))
					htProxyPool.remove(proxyElement.getHost());
				
				logger.warn("interrupted while waiting free Proxy.");
			}
		}else{
			logger.warn("this is not proxy element.");
		}
	}
	
	/* timer，some time update
	 * 
	 * @param 	int 	spaceTime
	 * @return void
	 */
	private void timerControl(int spaceTime)
	{
		isFetch = true;
			
		getProxyByDB();
		
		isFetch = false;
		
		Timer timer = new Timer(true);
		timer.schedule(new TimerTask(){
		   	public void run(){
		   		if(!isFetch){
		   			isFetch = true;
		   			
		   			getProxyByDB();
		   			updateInvalidProxy();
		   			
		   			isFetch = false;
		   		}
		   	}
		},spaceTime,spaceTime);
	}
	
	public void close()
	{
		
	}
	
	
	public int getProxyByDB()
	{
		int newAddCount = 0;
		
		Connection conn = null;
		Statement sm 	= null;
		ResultSet rs 	= null;
		
		StringBuilder sql = new StringBuilder(200);
		sql.append("SELECT ip, port, id, username, password FROM proxy WHERE status='Y' LIMIT ");
		sql.append(INT_PROXY_POOL_SIZE);
		
		try{
			conn 	= dbm.getConnection(DBTag);
			sm 		= conn.createStatement();
			rs 		= sm.executeQuery(sql.toString());
			
			ProxyElement proxy;
			while (rs.next()){
				if(!htProxyPool.containsKey(rs.getString(1))){
					InetSocketAddress isa = new InetSocketAddress(rs.getString(1), rs.getInt(2));
					
					proxy = new ProxyElement(Proxy.Type.HTTP, isa);

					proxy.setID(rs.getInt(3));
					proxy.setHost(rs.getString(1));
					proxy.setPort(rs.getInt(2));
					
					bkqProxyQueue.offer(proxy, INT_WAIT_TIMEOUT, TimeUnit.SECONDS);
					
					htProxyPool.put(rs.getString(1), proxy);
					
					newAddCount++;
				}
			}
			
		} catch (Exception e){
			logger.error("SQL : " + sql.toString());
			
			e.printStackTrace();
		}finally{
			if (sm != null){
	        	try{
	        		sm.close();
				} catch (Exception e){
				}
	        }
			
			if (conn != null)
				dbm.freeConnection(DBTag, conn);
        }
		
		logger.debug("fetch proxy data the end from DB. new add count " + newAddCount);
		
		return newAddCount;
	}

	public int updateInvalidProxy()
	{
		int updateCount 	= 0;
		
		ProxyElement proxy 	= null;
		
		Connection conn 	= null;
		Statement sm 		= null;
		
		StringBuilder sql = new StringBuilder(200);

		while(bkgInvalidProxyQueue.size() > 0){
			try{
				proxy = bkgInvalidProxyQueue.poll(INT_WAIT_TIMEOUT, TimeUnit.SECONDS);
				
				if(proxy != null){
					sql.delete(0, sql.length());
					sql.append("UPDATE proxy SET status='I' WHERE id=");
					sql.append(proxy.getID());
					
					conn 	= dbm.getConnection(DBTag);
					sm 		= conn.createStatement();
					
					updateCount += sm.executeUpdate(sql.toString());
				}
			} catch (Exception e){
				logger.error("SQL : " + sql.toString());
				
				e.printStackTrace();
			}finally{
				if (conn != null){
					dbm.freeConnection(DBTag, conn);
					
					conn = null;
				}
	        }
		}
		
		logger.debug("update invalid proxy status the end. update count : " + updateCount);
		
		return updateCount;
	}

}
